package com.cnebula.dataprocess.sqoop.job.adapter;

import java.util.List;
import java.util.UUID;

import org.apache.sqoop.client.SqoopClient;
import org.apache.sqoop.model.MLink;
import org.apache.sqoop.model.MLinkConfig;
import org.apache.sqoop.validation.Message;
import org.apache.sqoop.validation.Status;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.cnebula.dataprocess.sqoop.job.exception.SqoopMLinkCreateException;
import com.cnebula.dataprocess.sqoop.job.model.LinkConfig;
import com.cnebula.dataprocess.sqoop.job.model.LspMLink;

public class MLinkHDFSAdapter extends SqoopMLinkAdapter {

	private static MLinkHDFSAdapter instance;
	
	private static final Logger log = LoggerFactory.getLogger(MLinkHDFSAdapter.class);

	static {
		instance = new MLinkHDFSAdapter();
	}

	private MLinkHDFSAdapter(){}

	public static MLinkHDFSAdapter getInstance() {
		return instance;
	}

	@Override
	public LspMLink createMLink(SqoopClient client, LinkConfig config) throws SqoopMLinkCreateException {
		String linkName = "hdfs-link-"+ UUID.randomUUID().toString().substring(0, 10);
		MLink mlink = client.createLink("hdfs-connector");
		LspMLink link = new LspMLink(mlink);
		
		link.setName(linkName);
		link.setCreationUser("sqoop");
		link.setType(config.getType());
        link.setUrl(config.getUrl());
		
		MLinkConfig linkConfig = link.getConnectorLinkConfig();
		linkConfig.getStringInput("linkConfig.uri").setValue(config.getUrl());
		linkConfig.getStringInput("linkConfig.confDir").setValue(config.getHadoopConfDir());
		Status status = client.saveLink(link);
		if (status.canProceed()) {
			log.debug("Created Link with Link Name : " + link.getName());
            return link;
        } else {
        	List<Message> validationMessages = link.getValidationMessages();
        	String detailMsg = "";
        	for(Message msg : validationMessages){
        		if(!msg.getStatus().canProceed()){
        			detailMsg = msg.getMessage();
        			break;
        		}
        	}
        	throw new SqoopMLinkCreateException(detailMsg);
        }
	}

}
